package com.tour.dim

import com.shujia.common.{Constants, SparkTool}
import com.shujia.common.util.Md5
import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object UserTagMskApp extends SparkTool {
  /**
    *
    * 在子类中实现run方法，编写自定义的代码逻辑
    *
    * @param spark ： spark 入口
    */
  override def run(spark: SparkSession): Unit = {
    import spark.implicits._
    import org.apache.spark.sql.functions._


    //1、读取ods层总的用户画像表
    val odsUserTag: DataFrame = spark.sql(
      s"""
         |select * from
         |${Constants.ODS_DATABASE_NAME}.${Constants.USERTAG_TABLE_NAME}
         |where month_id=$month_id

      """.stripMargin)

    //使用自定义函数
    val md5Udf: UserDefinedFunction = udf((str: String) => Md5.md5(str))

    //2、对铭感信息加密
    val dimUserTagMsk: DataFrame = odsUserTag
      .select(
        md5Udf($"mdn") as "mdn",
        md5Udf($"name") as "name",
        $"gender",
        $"age",
        md5Udf($"id_number") as "id_number",
        $"number_attr",
        $"trmnl_brand",
        $"trmnl_price",
        $"packg",
        $"conpot",
        $"resi_grid_id",
        $"resi_county_id"
      )

    //保存数据
    dimUserTagMsk
      .write
      .format("csv")
      .mode(SaveMode.Overwrite)
      .option("sep", "\t")
      .save(s"${Constants.DIM_USERTAG_PATH_NAME}month_id=$month_id")

    //增加分区
    spark.sql(s"alter table ${Constants.DIM_DATABASE_NAME}.${Constants.DIM_USERTAG_TABLE_NAME} add if not exists partition(month_id='$month_id')")
  }
}
